# 流操作 stream
30 秒速记
- 流 = 数据一块一块地处理,不用一次性全读进内存,适合大文件、网络传输
- 四类:
Readable读、Writable写、Duplex双向、Transform边读边改 - 背压:
write()返回false就停,等drain再写;pipe自动帮你做 - 串流优先
pipeline():出错会传播并销毁整条链,pipe()不管错误 highWaterMark是缓冲水位不是上限:字节流按字节算,对象模式按个数算;Node 22起默认从16KB提到64KB
流就是把数据切成小块,一边生产一边消费,不用等全部数据到齐再处理。 好比用水管接水,而不是先把整池水搬过来。它的核心是背压:写入端处理不过来时 write() 会返回 false,读取端就该停一停,等 drain 事件再继续,这样内存才不会被撑爆。pipe() 会自动处理背压,但它不传播错误,中间一段挂了其他流不会关。所以实际项目里我串流一律用 pipeline(),错误能统一收到一个回调里,整条链也会被清理掉。
理解「流操作 stream」时,要先明确输入、运行阶段和可观察输出。这里的交互实验会分别执行正常路径与失败路径,并把 Node.js 的关键状态变化和同屏代码对应起来;不能只记命令名称。
工程中还要验证退出码、资源清理和异常输入。属于 stream 场景的配置应保持最小可复现,并在开发环境与 CI 中使用相同命令复跑,避免把平台差异或隐式默认值带到线上。
理解流,抓住「背压」和「错误处理」这两件事就够了。
背压是说生产者比消费者快的时候,要有机制让生产者停下来。Writable 的 write() 返回值就是这个信号:
const fs = require('fs');
const src = fs.createReadStream('big.log');
const dest = fs.createWriteStream('copy.log');
src.on('data', (chunk) => {
const ok = dest.write(chunk);
if (!ok) {
src.pause(); // 写入端缓冲满了,先停
dest.once('drain', () => src.resume()); // 缓冲排空再继续
}
});
src.on('end', () => dest.end());
这套逻辑 pipe() 已经帮你做了,但 pipe() 不管错误。生产环境推荐 stream/promises 里的 pipeline(Node 15 起可用):
const { pipeline } = require('stream/promises');
const zlib = require('zlib');
async function compress() {
try {
await pipeline(
fs.createReadStream('big.log'),
zlib.createGzip(),
fs.createWriteStream('big.log.gz'),
);
console.log('done');
} catch (err) {
console.error('failed:', err.message); // 任意一段出错都会到这里,且整条链已销毁
}
}
消费可读流的第三种方式是异步迭代器,for await (const chunk of readable) 会在你处理完当前块之后才拉下一块,背压是天然的。三种方式选一种用,不要在同一个流上混用。
面试官追问
追问 1用 fs.readFile 读一个 2GB 的日志文件,会发生什么?
要一次性把整个文件读进 Buffer,内存直接涨 2GB,超过 Buffer 上限还会直接报错。改成 fs.createReadStream 逐块处理,内存基本稳定在几十 KB。
追问 2a.pipe(b).pipe(c),b 报错了会怎样?
b 的错误不会传到 a 和 c,如果没监听 error 进程直接崩,监听了的话 a 和 c 也还挂着,文件句柄泄漏。改用 await pipeline(a, b, c),任何一段出错都会销毁整条链并抛出来。
追问 3手写 src.on('data', chunk => dest.write(chunk)),写入端慢的时候内存一直涨,为什么?
没处理背压,write() 返回 false 还在继续写,数据全堆在写入端的缓冲里。要在 false 时 src.pause(),等 dest 触发 drain 再 resume(),或者直接用 pipeline。
追问 4data 事件和 for await 能一起用吗?
不要混用,两种都会消费数据,混着用会抢数据或者丢数据。选一种:新代码推荐 for await (const chunk of stream),它自带背压,写法也像同步代码。
追问 5highWaterMark 调大是不是就更快?
不一定。它是缓冲水位,调大能减少暂停次数,但每个流占的内存也跟着涨,并发流多了反而危险。要不要调得看实测的吞吐和内存曲线。
# 模块概览
30 秒速记
- 很多核心对象本身就是流:
process.stdin/stdout、http的req/res、net.Socket、fs读写流 - 业务开发多数时候是「用流」,不用
require('stream');写自定义流才需要深入扩展API - 最小例子:
fs.createReadStream('./a.txt').pipe(process.stdout) - 流都继承自
EventEmitter,靠data/end/error/finish这些事件驱动 gulp、HTTP代理、文件上传下载,本质都是流的组合
流在 Node.js 里无处不在,你没写 require('stream') 不代表没在用它。 HTTP 服务里的 req 是可读流、res 是可写流,process.stdout 是可写流,net.Socket 是双工流。所以大部分人日常是在「用流」,需要搞懂的是它的运行机制:数据分块流动、靠事件驱动、有背压。只有像 gulp 插件那样要自己造一个流的时候,才需要去研究 Readable、Transform 这些类怎么继承实现。
nodejs的核心模块,基本上都是stream的的实例,比如process.stdout、http.clientRequest。
对于大部分的nodejs开发者来说,平常并不会直接用到stream模块,只需要了解stream的运行机制即可(非常重要)。
而对于想要实现自定义stream实例的开发者来说,就得好好研究stream的扩展API了,比如gulp的内部实现就大量用到了自定义的stream类型。
来个简单的例子镇楼,几行代码就实现了读取文件内容,并打印到控制台:
const fs = require('fs');
fs.createReadStream('./sample.txt').pipe(process.stdout);
面试官追问
追问 1HTTP 服务里要把一个大文件返回给前端,怎么写最省内存?
res 本身就是可写流,直接 await pipeline(fs.createReadStream(file), res)。别先 readFile 再 res.end(data),并发一多内存就扛不住。
追问 2fs.createReadStream('./a.txt') 在 CI 里报 ENOENT,本地正常,先看什么?
相对路径是相对 process.cwd() 解析的,不是相对当前文件。CI 的工作目录不同就找不到,改成 path.join(__dirname, 'a.txt')。
追问 3用户下载到一半关掉了页面,服务端的读文件流会自动停吗?
用 pipe 的话不会,读取流可能还挂着。pipeline 会在 res 被销毁时把源流一起销毁,这也是我推荐它的原因之一。
追问 4Express 里怎么拿到上传文件而不把整个文件读进内存?
req 就是可读流,用 busboy 这类库解析 multipart,拿到每个文件的流直接 pipeline 到磁盘或对象存储。body-parser 之类会把整个请求体读进内存,不适合大文件。
# Stream分类
30 秒速记
Readable:只出数据,fs.createReadStream()、http的req、process.stdinWritable:只收数据,fs.createWriteStream()、http的res、process.stdoutDuplex:能读能写,两个方向互不相干,典型是net.SocketTransform:特殊的Duplex,读出来的是写进去的加工结果,zlib、crypto- 判断
Duplex还是Transform:看输出是不是由输入算出来的
Node.js 的流分四类,记住「数据往哪走」就好分:只出的是 Readable,只进的是 Writable,进出都有的是 Duplex,进去的数据加工后再出来的是 Transform。 Duplex 和 Transform 最容易混。net.Socket 是 Duplex
